package org.tigr.htc.server.condor;

/*
This library is free software; you can redistribute it and/or
modify it under the terms of the GNU Lesser General Public
License as published by the Free Software Foundation; either
version 2.1 of the License, or (at your option) any later version.

This library is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
Lesser General Public License for more details.

You should have received a copy of the GNU Lesser General Public
License along with this library; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
*/

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.PrintWriter;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.text.DateFormat;
import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.log4j.Logger;
import org.tigr.antware.shared.command.events.CommandInterrupt;
import org.tigr.antware.shared.util.StringUtils;
import org.tigr.htc.cmd.Command;
import org.tigr.htc.cmd.CommandConfig;
import org.tigr.htc.cmd.ICommand;
import org.tigr.htc.cmd.Param;
import org.tigr.htc.cmd.ParamType;
import org.tigr.htc.common.ConfigUpdateLI;
import org.tigr.htc.common.HTCConfig;
import org.tigr.htc.server.IRunner;
import org.tigr.htc.server.RunnerMonitor;

/**
 * The <b>CondorRunner</b> class implements IRunner and is the runner that handles
 * the interface to Condor. CondorRunner takes a command and a working directory and then 
 * creates a Condor job. This involves:
 * 
 * 1. Creating a Condor submit file based on the Command and the CommandConfig
 * 2. Submitting the job to the correct queue
 * 3. Monitoring the job
 * 4. Sending events as the job progresses
 *
 */

public class CondorRunner implements IRunner, ConfigUpdateLI {
    static Logger log = Logger.getLogger(CondorRunner.class);

    /**
     * <code> hostIndex</code> pointer to the next SSH host
     */	
    protected static int hostIndex = 0;
    protected static String hostnames[] = HTCConfig.getProperty("machines").split(",");
	
    // regexp for parsing Condor logs
    static Pattern startEventPn = Pattern.compile(HTCConfig.getProperty("condor_start_event"));
    static Pattern endEventPn = Pattern.compile(HTCConfig.getProperty("condor_end_event"));
    static Pattern returnValuePn = Pattern.compile(HTCConfig.getProperty("condor_return_value"));
    static Pattern ipAddressPn = Pattern.compile(HTCConfig.getProperty("condor_ip_address"));

    // Condor requirement strings
    static final String LINUX = "((ARCH == \"INTEL\") && (OpSys == \"LINUX\"))";
    static final String OSF1 = "((ARCH == \"ALPHA\") && (OpSys == \"OSF1\"))";
    static final String SOLARIS = "((ARCH == \"SUN4u\") && (OpSys == \"SOLARIS28\") " 
	+ "|| (ARCH == \"SUN4u\") && (OpSys == \"SOLARIS29\") )";

    // member variables
    private Command cmd;
    private File cwd;
    private File eventLog;
    protected BufferedReader eventReader = null;
    protected int merrors = 0;

    /**
     * <code>CondorRunner</code> returns a new CondorRunner 
     *
     * @param pCmd a
     * @exception IllegalAccessException if set is attempted after submission
     */
    public CondorRunner(Command pCmd, File pCwd) {
		log.debug("Condor runner for " + pCmd.getID() + " construct using work dir " + pCwd);
		cmd = pCmd;
		cwd = new File(pCwd, "" + cmd.getID());
		
        // Set initial directory to 'cwd' if it is null
        log.debug("Command's initial directory is " + cmd.getInitialDir());
        if ( (cmd.getInitialDir() == null) || cmd.getInitialDir().equals("/")) {
            cmd.setInitialDir(cwd.getAbsolutePath());
        }
		
		eventLog = new File(cwd, "event.log");
    }

    public synchronized static int getNextHostIndex() {
		hostIndex++;
		if (hostIndex >= hostnames.length) {
			hostIndex = 0;
		}
		return hostIndex;
    }
    
    /* (non-Javadoc)
     * @see org.tigr.htc.server.IRunner#run()
     */
    public void run() {
    	HTCConfig.addConfigUpdateListener(this);

    	PrintWriter pw = null;

    	try {
    		log.debug("In run.");

    		// EVENT: Submit
    		log.info("Sending submit event for Command " + cmd.getID());

    		cmd.getConfig().setLogLocation(cwd.getAbsolutePath());
    		cmd.taskSubmitted(cmd.getSubmitURL(), eventLog.getAbsolutePath());

    		// setup
    		// make working directory
    		// TODO: fix delete to delete all sub directories
    		if (cwd.exists()) {
    			log.warn("CWD exists, going to delete it.");
    			if (!cwd.delete()) {
    				log.error(" Couldn't delete working directory needed by server " + cwd);
    			}
    		}

    		if (!cwd.mkdirs()) {
    			log.error("Couldn't create the working directory needed by the server " + cwd);
    			cmd.fireFailureEvent(new Exception(
    					" Couldn't create the working directory needed by the server "
    					+ cwd));
    			cmd.updateCommand();
    			return;
    		}

    		// Create event.log so that it is owned by us
    		if (!eventLog.createNewFile()) {
    			log.warn("False returned for create new file on: "
    					+ eventLog.getAbsolutePath());
    		}

    		// create writer for submit file
    		File conSubmit = new File(cwd, "condor.submit");
    		pw = new PrintWriter(new FileWriter(conSubmit), true);

    		initTasks();

    		// create the submit file
    		createSubmitFile(pw);
    		log.debug("Flushing and closing the submit file.");
    		pw.flush();
    		pw.close();
    		
    		File submitLog = new File(cwd, "submit.log");
    		log.debug("Trying to make a submit log" + submitLog.getAbsolutePath());
    		if (!submitLog.createNewFile()) {
    			log.warn("False returned for create new file on: "
    					+ submitLog.getAbsolutePath());
    		}

    		int index = CondorRunner.getNextHostIndex();
    		int CID = CondorUtils.submit(conSubmit, cmd.getUserName(), hostnames[index], submitLog);

    		if (CID <= 0) {
    			log.warn("Error running condor submit");
    			log.warn(" fire failure notice");
    			cmd.fireFailureEvent(new Exception("Could not connect to any Condor queues to submit the job."));
    			cmd.updateCommand();
    			return;
    		}

    		cmd.setQueue(hostnames[index]);
    		cmd.setGridJobID("" + CID);

    		// Monitor the log file
    		eventReader = new BufferedReader(new FileReader(eventLog));

    		log.debug("Done submitting. Now just wait in monitor mode. CID is "
    				+ CID);

    		RunnerMonitor.register(this);
    	} catch (Exception e) {
    		log.error("Error " + e.getMessage(), e);
    		cmd.fireFailureEvent(new Exception("Runner:" + e.getMessage()));
    		cmd.updateCommand();
    		if (eventReader != null) {
    			try {
    				eventReader.close();
    			} catch (java.io.IOException ioe) {
    				log.warn(" couldn't close event Reader");
    			}
    		}
    		return;
    	} finally {
    		// Always close the Condor submit file
    		if (pw != null) {
    			pw.close();
    		}
    		HTCConfig.removeConfigUpdateListener(this);
    	}
    }

	public void initTasks() {
		int tasks = cmd.getTimes();
		cmd.setNumTasks(tasks);
		cmd.getStatus().setWaiting(tasks);
	}

	public void createSubmitFile(PrintWriter pOut) {
		log.debug("In createSubmitFile.");
		pOut.println("# generated Condor submit file for Command "
				+ cmd.getID());
		pOut.println("# submitted for user " + cmd.getUserName());
		pOut.println("universe = vanilla");
		pOut.println("executable = " + cmd.getCommand());

		Iterator params = cmd.getParams().iterator();
		if (cmd.getParamsCount() > 0) {
			pOut.print("arguments = ");
		}

		for (; params.hasNext();) {
			Param param = (Param) params.next();

			if (param.getKey() != null) {
				pOut.print(param.getKey().replaceAll("\\$\\(Index\\)",
						"\\$\\(Process\\)")
						+ "=");
			}
			if (param.getType() == ParamType.PARAM) {
				pOut.print(param.getValue().replaceAll("\\$\\(Index\\)",
						"\\$\\(Process\\)")
						+ " ");
			} else {
				log.warn("Illegal/unknown parameter type for Command type htc.");
			}
		}

		if (cmd.getParamsCount() > 0) {
			pOut.println("");
		}

		if (cmd.getInitialDir() != null)
			pOut.println("initialdir = " + cmd.getInitialDir());

		if (cmd.getOutput() != null)
			pOut.println("output = "
					+ cmd.getOutput().replaceAll("$(Index)", "$(Process)"));

		if (cmd.getError() != null)
			pOut.println("error = "
					+ cmd.getError().replaceAll("$(Index)", "$(Process)"));

		if (cmd.getInput() != null)
			pOut.println("input = "
					+ cmd.getInput().replaceAll("$(Index)", "$(Process)"));

		pOut.println("# config variables");

		CommandConfig config = cmd.getConfig();
		if (config != null) {
			String requirements = "requirements = ";

			// if there are any hosts then override operating system
			if (config.getHosts() != null) {
				String[] hosts = config.getHosts().split(",");
				requirements += "( (machine == \"" + hosts[0] + "\")";
				for (int i = 1; i < hosts.length; i++) {
					requirements += " || (machine == \"" + hosts[i] + "\")";
				}
				requirements += " )";
			}

			if (config.getOpSys() != null) {
				String[] ops = config.getOpSys().split(",");
				log.debug("split opsys and length is " + ops.length);

				if (!requirements.equals("requirements = ")) {
					requirements += " && ";
				}

				for (int i = 0; i < ops.length; i++) {
					if (i != 0) {
						requirements += " || ";
					} else {
						requirements += " (";
					}

					if (ops[i].toLowerCase().equals("solaris")) {
						requirements += CondorRunner.SOLARIS;
					} else if (ops[i].toLowerCase().equals("osf1")) {
						requirements += CondorRunner.OSF1;
					} else if (ops[i].toLowerCase().equals("linux")) {
						requirements += CondorRunner.LINUX;
					}

				}
				requirements += ") ";
			}

			if (config.getMemory() >= 0) {
				requirements += " && ((Memory >= " + config.getMemory() + " ))";
			} else {
				// HACK to keep Condor from resetting memory size after eviction
				requirements += " && ((Memory >= " + 1 + " ))";
			}

			if ((config.getPassThrough() != null)
					&& (!config.getPassThrough().equals(""))) {
				requirements += " && (" + config.getPassThrough() + ")";
			}

			pOut.println("# requirements");
			pOut.println(requirements);

			// make rank string depends on group and length
			if (config.getLength() != null && (!config.getLength().equals("")))
				pOut.println("+Length = " + config.getLength());

			// output priority
			pOut.println("priority = " + config.getPriority());

			if (config.isGetenv()) {
				if (cmd.getEnv() != null) {
			    	Map<String, String> envMap = cmd.getEnv();
			    	
					pOut.print("environment = ");

			    	for (String envVar : envMap.keySet()) {
						pOut.print(envVar + " = " + envMap.get(envVar)+ ";");
					}
				} else {
					pOut.println("getenv = " + config.isGetenv());
				}
			}
		}
		pOut.println();
		// periodic remove based on length attribute
		pOut.println("notification = Never");
		pOut.println("log = " + eventLog.getAbsolutePath());
		pOut.println();

		pOut.println("queue " + cmd.getTimes());
		pOut.println(" # end of submit file ");
	}

	/**
	 * 
	 * @param p_line
	 *            line from a condor log file
	 * 
	 * 000 job submitted 001 job executing 002 job not executable 004 evicted
	 * 005 job terminated 007 shadow exception 009 job aborted by user 012 job
	 * was held 013 job was released log line example 005 (018.001.000) 08/21
	 * 11:53:28 Job terminated. event (cid.job.???) date time event info
	 * ^(\d+)\s\((\d+)\.(\d+)\.\d+\)\s(\d+)/(\d+)\s 1 2 3
	 * Pattern.compile("^([0-9]+)\\s+\\(([0-9]+)\\.([0-9]+)\\.[0-9]+\\)\\s(.*)");
	 * body ... end of event
	 */
	public boolean isRunnerComplete() {
		int type;
		int tcid;
		int jobIndex = 0;
		String strDate;
		String eventInfo;
		int returnValue;

		log.debug("Monitor job " + cmd.getID());
		String line;

		// while(true) {
		try {
			while ((line = eventReader.readLine()) != null) {
				try {
					// parse line
					log.debug("try to find a match " + line);

					Matcher matcher = startEventPn.matcher(line);
					if (matcher.matches()) {
						type = Integer.parseInt(matcher.group(1));
						tcid = Integer.parseInt(matcher.group(2));
						jobIndex = Integer.parseInt(matcher.group(3));
						strDate = matcher.group(4);
						eventInfo = matcher.group(5);

						log.debug(" --- event is " + type);
						log.debug("     cid is " + tcid);
						log.debug("     job number is " + jobIndex);
						log.debug("     date = " + strDate);
						log.debug("     eventInfo is " + eventInfo);

						DateFormat df = DateFormat.getDateTimeInstance(
								DateFormat.SHORT, DateFormat.SHORT);

						// Date eventDate = df.parse(strDate);
						Date eventDate = new Date();
						log.debug("**** test " + df.format(eventDate));
						// convert the Condor event to an HTC event
						if (type == 0) {
							// submit event
							Matcher ipMatch = ipAddressPn.matcher(eventInfo);
							if (ipMatch.matches()) {
								log.info("Matched " + ipMatch.group(1));
								// TODO check cmd state is WAITING
								log.debug("Got submit event from Condor for id "
												+ cmd.getID());
							}

						} else if (type == 1) {
							// start event
							Matcher ipMatch = ipAddressPn.matcher(eventInfo);
							if (ipMatch.matches()) {
								String IPAddress = ipMatch.group(1);
								log
										.debug("Trying reverse lookup of address >>"
												+ IPAddress + "<<");
								try {
									String machine = InetAddress.getByName(
											IPAddress).getHostName();
									log.debug("Lookup up returned " + machine);
									cmd.taskStarted(eventInfo, eventDate,
											machine, jobIndex, jobIndex);
								} catch (UnknownHostException e) {
									log.error("Reverse lookup failed "
											+ e.getMessage() + "\n");
									cmd.taskStarted(eventInfo, eventDate,
											IPAddress, jobIndex, jobIndex);
								}

							} else {
								log.error("Failed to match IP address.");
								cmd.taskStarted(eventInfo, eventDate,
										"Uknown host", jobIndex, jobIndex);
							}

						} else if (type == 5) {
							// finish event
							log.info("Event was a finish event");
							returnValue = parseReturnValue(eventReader);
							cmd.taskFinished(eventInfo, eventDate, returnValue,
									jobIndex);

							if (cmd.isFinished()) {
								if (cmd.getNotifyScript() != null) {
									notify(eventInfo, returnValue);
								}

								if (eventReader != null) {
									try {
										eventReader.close();
									} catch (java.io.IOException e) {
										log.warn(" couldn't close reader ");
									}
								}

								return true;
							}

						} else if (type == 9) {
							log.info("Condor runner firing interrupt event");
							cmd.taskInterrupted(eventInfo,
									CommandInterrupt.USER_KILL, jobIndex);
							if (cmd.isFinished()) {
								if (cmd.getNotifyScript() != null) {
									notify(eventInfo, -1);
								}

								if (eventReader != null) {
									try {
										eventReader.close();
									} catch (java.io.IOException e) {
										log.warn("Couldn't close reader ");
									}
								}

								return true;
							}

						} else if (type == 2) {
							log.info("Firing failure event");
							log.debug("Executable is invalid.");
							cmd.taskFailure(new Exception(
									"Job is not executable"), jobIndex);
							if (cmd.isFinished()) {
								if (cmd.getNotifyScript() != null) {
									notify(eventInfo, -1);
								}

								if (eventReader != null) {
									try {
										eventReader.close();
									} catch (java.io.IOException e) {
										log.warn("Couldn't close reader.");
									}
								}

								return true;
							}

						} else if (type == 4 || type == 12) {
							// evict event
							log.debug("Task is suspended");
						} else if (type == 13) {
							// resume event
						} else {
							log.warn("***** event ignored  *****");
							log.warn(" --- event is " + type);
							log.warn("     cid is " + tcid);
							log.warn("     job number is " + jobIndex);
							log.warn("     date = " + strDate);
							log.warn("     eventInfo is " + eventInfo);
						}

					}
				} catch (Exception e) {
					log.warn("logline is " + line, e);
				}
				log.debug("--- end of monitor while loop going to try to read another line ");
			}

			log.debug("No more lines to read in event.log");
			return false;
		} catch (Exception e) {
			log.warn("Error " + e.getMessage());
			merrors++;
			if (merrors > 3) {
				cmd.fireFailureEvent(new Exception(
						"Error in monitoring the command on the grid"));
				cmd.updateCommand();
				return true;
			}
			return false;
		}
	}

	/**
	 * @param eventReader
	 * @return
	 */
	public int parseReturnValue(BufferedReader eventReader) {
		log.debug("In parseReturnValue: " + cmd.getID());
		String line;

		try {
			while ((line = eventReader.readLine()) != null) {
				try {
					// parse line
					log.debug("Try to find return value");

					Matcher matcher = returnValuePn.matcher(line);

					if (matcher.matches()) {
						log.debug("Found return value ");
						return Integer.parseInt(matcher.group(1));
					} else {
						log.info("No match on return value ");
						return 255;
					}

				} catch (Exception e) {
					log.debug("In parseReturnValue logline is " + line, e);
					return 255;
				}
			}
			return 255;
		} catch (Exception e) {
			log.debug("Exception in finish event ", e);
			return 255;
		}

	}

	// method to run notify script and/or email user
	public void notify(String p_message, int p_returnValue) {
		log.debug("In notify.");

		if (cmd.getNotifyScript() == null) {
			log.warn("no script set ");
			return;
		}

		String email = cmd.getEmail();
		if (email == null) {
			email = cmd.getUserName();
		}

		File scriptFile = new File(cmd.getNotifyScript());
		String cid = "-requestID=" + cmd.getID();
		String returnVal = "-returnValue=" + p_returnValue;
		String lmessage = "-message=" + p_message;
		String emailArg = "-email=" + email;
		Process p;

		if (scriptFile.exists()) {
			log.debug("Notify script file exists.");
			try {
				String tmp = scriptFile.getAbsolutePath();
				if (log.isDebugEnabled()) {
					log.debug("Notify command is " + tmp);			
	                List args = Arrays.asList(new String[] { cid, returnVal, lmessage, emailArg, " -- " });
	                log.debug("Arguments: " + StringUtils.joinList(args, " "));
				}
				p = Runtime.getRuntime().exec(
						new String[] { scriptFile.getAbsolutePath(), cid,
								returnVal, lmessage, emailArg });
				int exitVal = p.waitFor();
				log.debug("Process exit value: " + exitVal);
			} catch (Exception e) {
				log.debug("Exception running notify scripts " + e.getMessage(), e);
			}
		} else {
			log.debug("Notify script was set but didn't exist: " + cmd.getNotifyScript());
		}
	}

	public ICommand getICommand() {
		return cmd;
	}

	public String getEventLog() {
		return eventLog.getAbsolutePath();
	}

	public boolean stop() throws Exception {
		// run GridUtils.condor_rm
		log.info("Stopping cmd " + cmd.getID() + " grid id = "
				+ cmd.getGridJobID());

		boolean result = CondorUtils.remove(Integer.parseInt(cmd.getGridJobID()),
				cmd.getQueue());
		// interrupt whole command
		return result;
	}

	public boolean hold() throws Exception {
		// run GridUtils.condor_hold
		return CondorUtils.hold(Integer.parseInt(cmd.getGridJobID()), cmd
				.getQueue());
	}

	public boolean release() throws Exception {
		// GridUtils.condor_release
		return CondorUtils.release(Integer.parseInt(cmd.getGridJobID()), cmd
				.getQueue());
	}

	public void configUpdated() {
		log.debug("Got a new config update... Reloading.");
		hostnames = HTCConfig.getProperty("machines").split(",");
		// Regular expressions for parsing Condor logs
		startEventPn = Pattern.compile(HTCConfig
				.getProperty("condor_start_event"));
		endEventPn = Pattern.compile(HTCConfig.getProperty("condor_end_event"));
		returnValuePn = Pattern.compile(HTCConfig
				.getProperty("condor_return_value"));
		ipAddressPn = Pattern.compile(HTCConfig
				.getProperty("condor_ip_address"));
	}

}

